package com.demo.pong.service.impl;

import com.demo.pong.exception.CustomException;
import com.demo.pong.orm.entity.ErrorLogs;
import com.demo.pong.orm.enumeration.ErrorLevelEnum;
import com.demo.pong.service.ErrorLogsService;
import com.demo.pong.service.PongService;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import io.micrometer.common.util.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.bson.Document;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Service;
import org.springframework.web.server.ResponseStatusException;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.Date;

/**
 * A specific service class that implements the PongService interface.
 * Provides the ability to respond to messages and handles distributed locks to prevent excessive requests.
 *
 * @author Jofo
 */
@Slf4j
@Service
public class PongServiceImpl implements PongService {
    private final ErrorLogsService errorLogsService;
    private final MongoDatabase mongoDatabase;

    /**
     * Constructors, which inject dependent services and databases.
     *
     * @param errorLogsService Error Log Service
     * @param mongoDatabase    MongoDB DB instance
     */
    public PongServiceImpl(ErrorLogsService errorLogsService, MongoDatabase mongoDatabase) {
        this.errorLogsService = errorLogsService;
        this.mongoDatabase = mongoDatabase;
    }

    // The name of the distributed lock collection
    private static final String LOCK_COLLECTION = "locks";
    // A unique identifier for a lock
    private static final String LOCK_ID = "pongServiceLock";
    // The lock timeout period, in milliseconds
    private static final long LOCK_TIMEOUT = 1000;

    /**
     * Respond to the client's message.
     * If the message is empty or does not equal "hello", an error is returned.
     * Attempts to acquire a distributed lock, and if successful, returns "World" after a delay of 100 ms.
     * If the lock cannot be acquired, an error log is logged and an error is returned with "Service is currently busy, please try again later."
     *
     * Messages sent @param msg client
     * @param traceId of the request
     * @return Mono object in response
     */
    @Override
    public Mono<String> respond(String msg, String traceId) {
        if (StringUtils.isBlank(msg)) {
            return Mono.error(new CustomException("msg is required"));
        }
        if (!"hello".equals(msg)) {
            return Mono.error(new ResponseStatusException(HttpStatus.BAD_REQUEST, HttpStatus.BAD_REQUEST.getReasonPhrase()));
        }

        // An attempt was made to obtain a distributed lock
        if (acquireLock()) {
            try {
                return Mono.just("World")
                        .delayElement(Duration.ofMillis(500));
            } finally {
                releaseLock();
            }
        } else {
            log.error(HttpStatus.TOO_MANY_REQUESTS.getReasonPhrase());
            ErrorLogs errorLog = new ErrorLogs();
            errorLog.setErrorLevel(ErrorLevelEnum.MEDIUM.getMessage());
            errorLog.setErrorCode(HttpStatus.TOO_MANY_REQUESTS.value());
            errorLog.setTraceId(traceId);
            errorLog.setStackTrace("PongServiceImpl respond");
            errorLog.setErrorMessage("Pong Service is currently busy, please try again later.");
            errorLogsService.saveErrorLog(errorLog);
            return Mono.error(new ResponseStatusException(HttpStatus.TOO_MANY_REQUESTS, "Service is currently busy, please try again later."));
        }
    }

    /**
     * Obtain distributed locks.
     * Try to insert a lock document, if the insertion is successful, it means that the lock was successfully acquired.
     * If the insertion fails, check if the existing lock has expired, and if it expires, release the lock and try to get it again.
     *
     * @return Whether the lock was successfully acquired
     */
    private boolean acquireLock() {
        MongoCollection<Document> collection = mongoDatabase.getCollection(LOCK_COLLECTION);
        Document lockDoc = new Document("_id", LOCK_ID)
                .append("locked", true)
                .append("timestamp", new Date());

        // Try inserting a lock document
        try {
            collection.insertOne(lockDoc);
            // The lock was successfully obtained
            return true;
        } catch (Exception e) {
            // Check if the existing lock has expired
            Document existingLock = collection.find(new Document("_id", LOCK_ID)).first();
            if (existingLock != null) {
                Date lockTime = existingLock.getDate("timestamp");
                long elapsedTime = new Date().getTime() - lockTime.getTime();
                log.info("Existing lock found, elapsed time: {} ms", elapsedTime);
                if (elapsedTime > LOCK_TIMEOUT) {
                    // The lock has expired, release the lock
                    releaseLock();
                    return true;
                }
            }
            log.warn("Failed to acquire lock: {}", e.getMessage());
            // Failed to acquire locks
            return false;
        }
    }

    /**
     * Release the distributed lock.
     * Remove lock documents.
     */
    private void releaseLock() {
        MongoCollection<Document> collection = mongoDatabase.getCollection(LOCK_COLLECTION);
        collection.deleteOne(new Document("_id", LOCK_ID));
    }
}
